package cn.com.sparkle.paxos.deamon;

import java.util.Set;

import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.ClusterState;
import cn.com.sparkle.paxos.NodeState;
import cn.com.sparkle.paxos.NodesCollection;
import cn.com.sparkle.paxos.event.EventsManager;
import cn.com.sparkle.paxos.event.NodeStateChangeEvent;
//import cn.com.sparkle.paxos.model.ElectionId;
import cn.com.sparkle.paxos.net.CallBack;
import cn.com.sparkle.paxos.net.MajorSender;
import cn.com.sparkle.paxos.net.NetNode;
//import cn.com.sparkle.paxos.net.message.SenatorHeatBeatResponse;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.ElectionId;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.Id;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.MessagePackage;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.SenatorHeatBeatRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.SenatorHeatBeatResponse;

public class HeatBeatCheckDeamon extends Thread {
	private final static Logger logger = Logger
			.getLogger(HeatBeatCheckDeamon.class);
	public final static int HEAT_BEAT_INTERVAL = 10000;
	private ClusterState cState;
	private EventsManager eventsManager;

	private class HeatBeatCallBack implements CallBack {
		@Override
		public void call(NetNode nnode, MessagePackage value) {
			if (value.hasSenatorHeartBeatResponse()) {
				SenatorHeatBeatResponse res = value
						.getSenatorHeartBeatResponse();
				NodeState nodeState = nnode.getNodeState();
				nodeState.setLastBeatHeatTime(System.currentTimeMillis());
				nodeState.setMasterConnected(res.getIsMasterConnected());
				// nodeState.setLastElectionId(new
				// ElectionId(res.getElectionId(), res.getElectionAddress(),
				// res.getElectionVersion()));
				nodeState.setLastElectionId(ElectionId
						.newBuilder()
						.setId(Id.newBuilder()
								.setIncreaseId(res.getElectionId())
								.setAddress(res.getElectionAddress()))
						.setVersion(res.getElectionVersion()).build());
				nodeState.setLastCanExecuteInstanceId(res
						.getLastCanExecuteInstanceId());
				NodeStateChangeEvent.doBeatHeartEvent(eventsManager, nodeState);
			} else {
				throw new RuntimeException("unexception error:no response!");
			}
		}

		@Override
		public void fail(NetNode nnode) {
		}
	}

	public HeatBeatCheckDeamon(ClusterState cState, EventsManager eventsManager) {
		this.cState = cState;
		this.eventsManager = eventsManager;
		this.setName("Paxos-HeatBeat-Checker");
	}

	@Override
	public void run() {
		CallBack heatBeatCallBack = new HeatBeatCallBack();
		while (true) {
			// just senator to senator
			NodesCollection senators = cState.getSenators();
			Set<NetNode> activeSenator = senators.getActiveNodes();
			SenatorHeatBeatRequest request = PaxosMessages.SenatorHeatBeatRequest
					.getDefaultInstance();
			logger.debug("active senator num:" + activeSenator.size());
			MajorSender.sendToMajor(activeSenator, request, heatBeatCallBack);
			try {
				Thread.sleep(HEAT_BEAT_INTERVAL);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

}
